package com.sjsu.aws.db;

/**
 * 
 * @author Deepali Pawar
 *
 */
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.dynamodb.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodb.model.AttributeAction;
import com.amazonaws.services.dynamodb.model.AttributeValue;
import com.amazonaws.services.dynamodb.model.AttributeValueUpdate;
import com.amazonaws.services.dynamodb.model.CreateTableRequest;
import com.amazonaws.services.dynamodb.model.DeleteItemRequest;
import com.amazonaws.services.dynamodb.model.DeleteItemResult;
import com.amazonaws.services.dynamodb.model.DeleteTableRequest;
import com.amazonaws.services.dynamodb.model.DeleteTableResult;
import com.amazonaws.services.dynamodb.model.DescribeTableRequest;
import com.amazonaws.services.dynamodb.model.Key;
import com.amazonaws.services.dynamodb.model.KeySchema;
import com.amazonaws.services.dynamodb.model.KeySchemaElement;
import com.amazonaws.services.dynamodb.model.ListTablesResult;
import com.amazonaws.services.dynamodb.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodb.model.PutItemRequest;
import com.amazonaws.services.dynamodb.model.PutItemResult;
import com.amazonaws.services.dynamodb.model.ReturnValue;
import com.amazonaws.services.dynamodb.model.ScanRequest;
import com.amazonaws.services.dynamodb.model.ScanResult;
import com.amazonaws.services.dynamodb.model.TableDescription;
import com.amazonaws.services.dynamodb.model.TableStatus;
import com.amazonaws.services.dynamodb.model.UpdateItemRequest;
import com.amazonaws.services.dynamodb.model.UpdateItemResult;
import com.amazonaws.services.dynamodb.model.UpdateTableRequest;
import com.amazonaws.services.dynamodb.model.UpdateTableResult;

public class TableOperationImpl implements TableOperation {
	
	private final static Logger	          LOG	= Logger
	                                                  .getLogger(TableOperationImpl.class);
	protected static AmazonDynamoDBClient	dynamoDB;
	
	protected void init() throws Exception {
		
		AWSPropertyUtil.loadLogFile();
		
		LOG.info("/resources/AWSCreddentials");
		
		String awsPath = TableOperationImpl.class.getResource("/").getPath()
		        + "../resources/AwsCredentials.properties";
		System.out.println("awsPath:" + awsPath);
		
		InputStream in = new FileInputStream(new File(awsPath));
		AWSCredentials credentials = new PropertiesCredentials(in);
		dynamoDB = new AmazonDynamoDBClient(credentials);
	}
	
	public TableOperationImpl() throws Exception {
		super();
		init();
	}
	
	@Override
	public boolean insertItem(String tableName, TableRow row) {
		Map<String, AttributeValue> item = row.getItem();
		PutItemRequest putItemRequest = new PutItemRequest(tableName, item);
		PutItemResult putItemResult = dynamoDB.putItem(putItemRequest);
		LOG.info("PutItemResult" + putItemResult);
		return true;
	}
	
	@Override
	public boolean UpdatePutItem(String tableName, TableRow row) {
		
		Map<String, AttributeValueUpdate> updateItems = new HashMap<String, AttributeValueUpdate>();
		Key key = new Key().withHashKeyElement(new AttributeValue().withS(row
		        .getUpdateKey()));
		
		updateItems.put(
		        row.getUpdateColumnName(),
		        new AttributeValueUpdate().withAction(AttributeAction.PUT)
		                .withValue(
		                        new AttributeValue().withS(row
		                                .getUpdateColumnValue())));
		
		UpdateItemRequest updateItemRequest = new UpdateItemRequest()
		        .withTableName(tableName).withKey(key)
		        .withReturnValues(ReturnValue.UPDATED_NEW)
		        .withAttributeUpdates(updateItems);
		
		UpdateItemResult result = dynamoDB.updateItem(updateItemRequest);
		LOG.info("Result" + result);
		return true;
	}
	
	@Override
	public boolean UpdatePutItemWithN(String tableName, TableRow row) {
		
		Map<String, AttributeValueUpdate> updateItems = new HashMap<String, AttributeValueUpdate>();
		Key key = new Key().withHashKeyElement(new AttributeValue().withS(row
		        .getUpdateKey()));
		
		updateItems.put(
		        row.getUpdateColumnName(),
		        new AttributeValueUpdate().withAction(AttributeAction.PUT)
		                .withValue(
		                        new AttributeValue().withN(row
		                                .getUpdateColumnValue())));
		
		UpdateItemRequest updateItemRequest = new UpdateItemRequest()
		        .withTableName(tableName).withKey(key)
		        .withReturnValues(ReturnValue.UPDATED_NEW)
		        .withAttributeUpdates(updateItems);
		
		UpdateItemResult result = dynamoDB.updateItem(updateItemRequest);
		LOG.info("Result: " + result);
		
		return true;
	}
	
	@Override
	public boolean UpdateDeleteItem(String tableName, TableRow row) {
		
		Map<String, AttributeValueUpdate> updateItems = new HashMap<String, AttributeValueUpdate>();
		Key key = new Key().withHashKeyElement(new AttributeValue().withS(row
		        .getUpdateKey()));
		
		updateItems.put(
		        row.getUpdateColumnName(),
		        new AttributeValueUpdate().withAction(AttributeAction.DELETE)
		                .withValue(
		                        new AttributeValue().withS(row
		                                .getUpdateColumnValue())));
		
		UpdateItemRequest updateItemRequest = new UpdateItemRequest()
		        .withTableName(tableName).withKey(key)
		        .withReturnValues(ReturnValue.UPDATED_NEW)
		        .withAttributeUpdates(updateItems);
		
		UpdateItemResult result = dynamoDB.updateItem(updateItemRequest);
		LOG.info("Result: " + result);
		
		return true;
	}
	
	@Override
	public boolean UpdateTable(String tableName) {
		
		ProvisionedThroughput provisionedThroughput = new ProvisionedThroughput()
		        .withReadCapacityUnits(15L).withWriteCapacityUnits(12L);
		
		UpdateTableRequest updateTableRequest = new UpdateTableRequest()
		        .withTableName(tableName).withProvisionedThroughput(
		                provisionedThroughput);
		
		UpdateTableResult result = dynamoDB.updateTable(updateTableRequest);
		LOG.info("Updated" + result);
		
		return true;
	}
	
	@Override
	public boolean deleteItem(String tableName, TableRow row) {
		
		ReturnValue returnValues = ReturnValue.ALL_OLD;
		DeleteItemRequest deleteItemRequest = new DeleteItemRequest()
		        .withTableName(tableName)
		        .withKey(
		                new Key().withHashKeyElement(new AttributeValue()
		                        .withS(row.getDeleteKey())))
		        .withReturnValues(returnValues);
		DeleteItemResult result = dynamoDB.deleteItem(deleteItemRequest);
		Map<String, AttributeValue> item = result.getAttributes();
		if (item == null)
			//System.out.println("not deleted result" + result.toString());
		//else
			System.out.println("Deleted " + result.toString());
		
		return true;
	}
	
	@Override
	public boolean deleteItemWithCondition(String tableName, TableRow row) {
		
		ReturnValue returnValues = ReturnValue.ALL_OLD;
		DeleteItemRequest deleteItemRequest = new DeleteItemRequest()
		        .withTableName(tableName)
		        .withKey(
		                new Key().withHashKeyElement(new AttributeValue()
		                        .withS(row.getDeleteKey())))
		        .withExpected(row.getExpectedDeletedValues())
		        .withReturnValues(returnValues);
		
		DeleteItemResult result = dynamoDB.deleteItem(deleteItemRequest);
		Map<String, AttributeValue> item = result.getAttributes();
		if (item == null)
			System.out.println("not deleted result" + result.toString());
		else
			System.out.println("deleted result" + result.toString());
		
		return true;
	}
	
	@Override
	public boolean deleteItemWithConditionAndMultipleKeys(String tableName,
	        TableRow row) {
		
		ReturnValue returnValues = ReturnValue.ALL_OLD;
		DeleteItemRequest deleteItemRequest = new DeleteItemRequest()
		        .withTableName(tableName)
		        .withKey(
		                new Key().withHashKeyElement(
		                        new AttributeValue().withS(row.getDeleteKey()))
		                        .withRangeKeyElement(
		                                new AttributeValue().withS(row
		                                        .getDeleteKey1())))
		        .withExpected(row.getExpectedDeletedValues())
		        .withReturnValues(returnValues);
		
		DeleteItemResult result = dynamoDB.deleteItem(deleteItemRequest);
		Map<String, AttributeValue> item = result.getAttributes();
		if (item == null)
			System.out.println("not deleted result" + result.toString());
		else
			System.out.println("deleted result" + result.toString());
		
		return true;
	}
	
	@Override
	public boolean deleteTable(String tableName) {
		DeleteTableRequest deleteTableRequest = new DeleteTableRequest()
		        .withTableName(tableName);
		DeleteTableResult result = dynamoDB.deleteTable(deleteTableRequest);
		System.out.println("Deleted" + result);
		return true;
	}
	
	@Override
	public boolean createTable(String tableName, String keyName, String keyType) {
		// Create a table with a primary key named 'name', which holds a string
		CreateTableRequest createTableRequest = new CreateTableRequest()
		        .withTableName(tableName)
		        .withKeySchema(
		                new KeySchema(new KeySchemaElement().withAttributeName(
		                        keyName).withAttributeType(keyType)))
		        .withProvisionedThroughput(
		                new ProvisionedThroughput().withReadCapacityUnits(10L)
		                        .withWriteCapacityUnits(10L));
		TableDescription createdTableDescription = dynamoDB.createTable(
		        createTableRequest).getTableDescription();
		System.out.println("Created Table: " + createdTableDescription);
		
		// Wait for it to become active
		waitForTableToBecomeAvailable(tableName);
		
		// Describe our new table
		DescribeTableRequest describeTableRequest = new DescribeTableRequest()
		        .withTableName(tableName);
		TableDescription tableDescription = dynamoDB.describeTable(
		        describeTableRequest).getTable();
		LOG.info("Table Description: " + tableDescription);
		
		return true;
	}
	
	@Override
	public boolean createTableWithMultipleKeys(String tableName,
	        String keyName1, String keyType1, String keyName2, String keyType2) {
		KeySchemaElement groupSchemaElement = new KeySchemaElement()
		        .withAttributeName(keyName1).withAttributeType(keyType1)
		        .withAttributeName(keyName2).withAttributeType(keyType2);
		KeySchemaElement groupSchemaElement1 = new KeySchemaElement()
		        .withAttributeName(keyName1).withAttributeType(keyType1);
		
		KeySchema groupSchema = new KeySchema(groupSchemaElement1);
		groupSchema.setRangeKeyElement(groupSchemaElement);
		CreateTableRequest createTableRequest = new CreateTableRequest()
		        .withTableName(tableName)
		        .withKeySchema(groupSchema)
		        .withProvisionedThroughput(
		                new ProvisionedThroughput().withReadCapacityUnits(10L)
		                        .withWriteCapacityUnits(10L));
		TableDescription createdTableDescription = dynamoDB.createTable(
		        createTableRequest).getTableDescription();
		System.out.println("Created Table: " + createdTableDescription);
		
		// Wait for it to become active
		waitForTableToBecomeAvailable(tableName);
		
		// Describe our new table
		DescribeTableRequest describeTableRequest = new DescribeTableRequest()
		        .withTableName(tableName);
		TableDescription tableDescription = dynamoDB.describeTable(
		        describeTableRequest).getTable();
		System.out.println("Table Description: " + tableDescription);
		
		return true;
	}
	
	@Override
	public ScanResult queryTable(String tableName, QueryCriteria criteria) {
		
		ScanRequest scanRequest = new ScanRequest(tableName)
		        .withScanFilter(criteria.getScanFilter());
		ScanResult scanResult = dynamoDB.scan(scanRequest);
		// System.out.println("Result Scan: " + scanResult.getCount());
		
		return scanResult;
	}
	
	@Override
	public ScanResult queryTableWithNoCondition(String tableName) {
		
		ScanRequest scanRequest = new ScanRequest(tableName);
		ScanResult scanResult = dynamoDB.scan(scanRequest);
		// System.out.println("Result Scan: " + scanResult.getCount());
		return scanResult;
	}
	
	protected static void waitForTableToBecomeAvailable(String tableName) {
		System.out.println("Waiting for " + tableName + " to become ACTIVE...");
		
		long startTime = System.currentTimeMillis();
		long endTime = startTime + (10 * 60 * 1000);
		while (System.currentTimeMillis() < endTime) {
			try {
				Thread.sleep(1000 * 20);
			} catch (Exception e) {
			}
			try {
				DescribeTableRequest request = new DescribeTableRequest()
				        .withTableName(tableName);
				TableDescription tableDescription = dynamoDB.describeTable(
				        request).getTable();
				String tableStatus = tableDescription.getTableStatus();
				System.out.println("  - current state: " + tableStatus);
				if (tableStatus.equals(TableStatus.ACTIVE.toString()))
					return;
			} catch (AmazonServiceException ase) {
				if (ase.getErrorCode().equalsIgnoreCase(
				        "ResourceNotFoundException") == false)
					throw ase;
			}
		}
		
		throw new RuntimeException("Table " + tableName + " never went active");
	}
	
	public ListTablesResult listTables() {
		return dynamoDB.listTables();
	}
}
